package rhc.discribute.node.connecter.schedule;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;
import org.quartz.Job;
import org.quartz.JobDataMap;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;

import rhc.discribute.node.connecter.impl.ZKPathLockData;
import rhc.discribute.node.connecter.impl.zookper.ZKExecute;
import rhc.discribute.node.sendMessage.SendMessageCallback;
import rhc.discribute.node.sendMessage.SendMessageData;
import rhc.discribute.util.ScheduleUpdate;

/**一条消息发送后会持有一个分布式锁等待对方接收后才会释放锁，如果对方没有接收则会一直持有锁造成不能再发消息到该节点
 * 为了避免这种情况会在发送消息5秒后执行该检查任务，如果没有接收成功则主动释放，该类为检查的job
 * @author rhc
 *
 */
public class SendMessageFailCheckJob implements Job{

	private static final Logger logger=Logger.getLogger(SendMessageFailCheckJob.class);
	
	@Override
	public void execute(JobExecutionContext arg0) throws JobExecutionException {
		//在创建任务时会将本对象绑定到任务
		JobDataMap jobDataMap=arg0.getMergedJobDataMap();
		
		SendMessageFailCheckSchedule schedule=(SendMessageFailCheckSchedule) jobDataMap.get(ScheduleUpdate.SCHEDULE_UPDATE);
		
		SendMessageData dataMap=schedule.getDataMap();
		
		CountDownLatch latch=dataMap.getLatch();
		ZKExecute zkExecute=dataMap.getZkExecute();
		String messagePath=dataMap.getMessagePath();
		SendMessageCallback callback=dataMap.getCallback();
		ZKPathLockData zkLockData=dataMap.getZkLockData();
		
		try {
			//等待1毫秒如果锁还没被释放则主动释放，
			if(!latch.await(1, TimeUnit.MILLISECONDS)){
				latch.countDown();
				zkExecute.sendToHostMessageFail(messagePath,zkLockData,callback);
			}
		} catch (Exception e) {
			logger.error("发送消息到路径 "+messagePath+" 执行发送消息失败时出错，"+e.getMessage());
			e.printStackTrace();
		}finally{
			//zk对象是一个永久对象，主动删除引用
			dataMap.clear();
		}
	}

}
